package com.nulldev.util.concurrency.threadIt.v4.emapi.ft;

import static com.nulldev.util.concurrency.threadIt.v4.emapi.uft.ExecutableUtils.callAsyncEventsOrFail;
import static com.nulldev.util.concurrency.threadIt.v4.emapi.uft.ExecutableUtils.callAsyncFails;

import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

import com.nulldev.util.JVM.Arguments;
import com.nulldev.util.VariableAPI.MathUtil;
import com.nulldev.util.concurrency.threadIt.v3.extensions.Task;
import com.nulldev.util.concurrency.threadIt.v3.extensions.TaskStatus;
import com.nulldev.util.concurrency.threadIt.v4.asyncTasks.AsyncTaskCallback;
import com.nulldev.util.concurrency.threadIt.v4.emapi.ExecutorManager;
import com.nulldev.util.concurrency.threadIt.v4.emapi.uft.Executable;
import com.nulldev.util.concurrency.threadIt.v4.locking.ResettableCountDownLatch;
import com.nulldev.util.concurrency.threadIt.v4.utils.impls.LatchableSemaphore;
import com.nulldev.util.concurrency.threadIt.v4.utils.impls.TaskTimedOutException;
import com.nulldev.util.data.Variables;
import com.nulldev.util.data.Arrays.queues.EnhancedLinkedBlockingQueue;
import com.nulldev.util.internal.backport.concurrency9.Lists;
import com.nulldev.util.io.FastIO.FastIO;
import com.nulldev.util.manage.Version;
import com.nulldev.util.manage.legacy.Timer;

/**
 * FutureTask, the successor to the Task API from v3.
 * 
 * @author null
 * @api-version 4.0
 * @since 3.0-alpha
 * @param <T>
 */
@SuppressWarnings("deprecation")
public abstract class FutureTask<T> implements Runnable, Executable<T> {

	/**
	 * VERSION 1.0: (current) <br>
	 * - Initial release. <br>
	 * VERSION 1.1: <br>
	 * - Catch Throwable instead of Exception [DONE] <br>
	 * - EXPERIMENT: Use custom semaphores for waiting instead of CDLs. [DONE] <br>
	 * - Cancelable tasks. <br>
	 * - Slave tasks and task chaining. <br>
	 * - Polling API. <br>
	 * VERSION 1.2: <br>
	 * - Use atomic values for storing results and statuses. [DONE] <br>
	 * - Attempt to remove currentThread() <br>
	 * - Rewrite task cancel() and kill() methods. <br>
	 * - Default to offloaded async. [DONE] <br>
	 * - Default to new fail logic. [DONE]
	 */
	public static final double VERSION = 1.06;
	public static final Version LIB_VERSION = new Version("1.0.6");

	private final static boolean ENABLE_OFFLOADED_ASYNC = Arguments.notArguments("--Tasks.disableAsyncOffload", "--FutureTask.disableAsyncOffload");
	private final static boolean ENABLE_ELBQ = Arguments.hasArgument("--Tasks.enableELBQ");
	private final static boolean USE_LAZY_AWAIT = Arguments.notArguments("--Tasks.disableLazyAwait");
	/*
	 * FIXME: Events should NOT be chained and should execute with priority. Disable
	 * async-event-submission for now.
	 */
	private final static boolean ENABLE_ASYNC_EVENT_SUBMISSION = Arguments.hasArgument("--Tasks.enableAsyncEventSubmission");
	private final static boolean USE_NEW_FAIL_LOGIC = Arguments.notArguments("--Tasks.disableNewFailureLogic");

	/**
	 * Conclusion from experiment: <br>
	 * - CountdownLatches have much better latency. <br>
	 * - Stability is fine, mainly just EMAPI having usual bugs or signaling issues.
	 */
	private final static boolean USE_SEMAPHORE_EXPERIMENT = true;

	private final BlockingQueue<AsyncTaskCallback<T>> callback;
	private final AtomicReference<T> result = new AtomicReference<T>();
	private final AtomicReference<Throwable> cachedError = new AtomicReference<Throwable>();
	private final AtomicReference<TaskStatus> status = new AtomicReference<TaskStatus>(TaskStatus.UNSUBMITTED);
	private final AtomicReference<Thread> t = new AtomicReference<Thread>();
	private final Throwable e;

	private final ResettableCountDownLatch countDownLatch = USE_SEMAPHORE_EXPERIMENT ? null : new ResettableCountDownLatch(1);
	private final LatchableSemaphore sem = USE_SEMAPHORE_EXPERIMENT ? new LatchableSemaphore(0, /* fifo */ true) : null;

	public FutureTask() {
		this.callback = ENABLE_ELBQ ? new EnhancedLinkedBlockingQueue<AsyncTaskCallback<T>>() : new ArrayBlockingQueue<AsyncTaskCallback<T>>(128);
		this.e = new Exception("/// TRACE /// [" + MathUtil.toHex(super.hashCode()) + "]").fillInStackTrace();
	}

	public FutureTask(final BlockingQueue<AsyncTaskCallback<T>> callback) {
		this.callback = callback;
		this.e = new Exception("\\\\\\ TRACE \\\\\\  [" + MathUtil.toHex(super.hashCode()) + "]").fillInStackTrace();
	}

	@Override
	public void run(final Thread currentThread) {
		/* We should be queued by now, otherwise something clearly went wrong. */
		if (!this.status.weakCompareAndSet(TaskStatus.QUEUED, TaskStatus.RUNNING)) {
			new RuntimeException("[FutureTask::run] weakCompareAndSet failed()!!! (" + this.status + ")").printStackTrace();
			this.getTrace().printStackTrace();

			rejectSubmission();
			return;
		}

		this.t.lazySet(currentThread);

		try {
			final T r = execute();
			this.result.lazySet(r);
			this.status.set(TaskStatus.COMPLETED);

			/*
			 * Give events lower priority over those directly waiting using
			 * waitTillCompletion(), etc...
			 */
			if (USE_SEMAPHORE_EXPERIMENT) {
				this.sem.release(Integer.MAX_VALUE);
			} else {
				this.countDownLatch.countDown();
			}

			callAsyncEventsOrFail(ENABLE_OFFLOADED_ASYNC, ENABLE_ASYNC_EVENT_SUBMISSION, this.callback, r);
		} catch (Throwable e) {
			this.status.set(TaskStatus.FAILED);
			this.cachedError.lazySet(e);

			/*
			 * See above comment.
			 */
			if (USE_SEMAPHORE_EXPERIMENT) {
				this.sem.release(Integer.MAX_VALUE);
			} else {
				this.countDownLatch.countDown();
			}

			callAsyncFails(ENABLE_OFFLOADED_ASYNC, ENABLE_ASYNC_EVENT_SUBMISSION, this.callback, e);
			if (!USE_NEW_FAIL_LOGIC)
				throw new RuntimeException(e);
		}
	}

	@Override
	public void run() {
		this.run(null);
	}

	public abstract T execute();

	/**
	 * Freezes the current thread until either the following occur: <br>
	 * - The task completes and returns the result. <br>
	 * - The task fails or is killed and <code>NULL</code> is returned. (old logic)
	 * <br>
	 * - The task fails and an exception is thrown. (new logic)
	 */
	public T waitTillCompletion() {
		if ((USE_SEMAPHORE_EXPERIMENT ? this.completed() : this.countDownLatch.isCompleted())) {
			final TaskStatus _status = this.status.get();

			if (_status == TaskStatus.FAILED && USE_NEW_FAIL_LOGIC) {
				throw new RuntimeException(this.cachedError.get());
			} else if (_status == TaskStatus.SUBMISSION_FAILED && USE_NEW_FAIL_LOGIC) {
				throw new RuntimeException("Submission was rejected.");
			} else {
				return this.result.get();
			}

		} else {
			try {
				if (USE_SEMAPHORE_EXPERIMENT) {
					this.sem.acquire(1);
				} else {
					if (USE_LAZY_AWAIT) {
						this.countDownLatch.lazyAwait();
					} else {
						this.countDownLatch.await();
					}
				}
			} catch (InterruptedException _ignored) {
			}

			final TaskStatus _status = this.status.get();

			if (_status == TaskStatus.FAILED && USE_NEW_FAIL_LOGIC) {
				throw new RuntimeException(this.cachedError.get());
			} else if (_status == TaskStatus.SUBMISSION_FAILED && USE_NEW_FAIL_LOGIC) {
				throw new RuntimeException("Submission was rejected.");
			} else {
				return this.result.get();
			}
		}
	}

	@Override
	public boolean completed() {
		final TaskStatus _status = this.status.get();
		return _status.isCompleted();
	}

	@Override
	public TaskStatus status() {
		return this.status.get();
	}

	@Override
	public T result() {
		return this.result.get();
	}

	/**
	 * Attempts to kill the executing thread.
	 */
	public void kill() {
		this.status.set(TaskStatus.KILLED);
		if (this.t.get() != null) {
			this.t.get().interrupt();
		}
		try {
			if (USE_SEMAPHORE_EXPERIMENT) {
				this.sem.release(Integer.MAX_VALUE);
			} else {
				this.countDownLatch.countDown();
			}
		} catch (Exception e) {
		}
	}

	public boolean addCallbackTask(final AsyncTaskCallback<T> task) {
		if (task == null) {
			return false;
		} else if ((USE_SEMAPHORE_EXPERIMENT ? this.completed() : this.countDownLatch.isCompleted())) {
			final TaskStatus _status = this.status.get();

			if (_status == TaskStatus.FAILED) {
				task.onError(this.cachedError.get());
			} else if (_status == TaskStatus.SUBMISSION_FAILED) {
				task.onSubmissionRejection();
			} else {
				try {
					task.onCompletion(this.result.get());
				} catch (Throwable tx) {
					task.onError(tx);
				}
			}
			return true;

		} else {

			try {
				return this.callback.offer(task);
			} catch (Throwable _internal) {

				if (this.completed()) {
					final TaskStatus _status = this.status.get();

					if (_status == TaskStatus.FAILED) {
						task.onError(this.cachedError.get());
					} else if (_status == TaskStatus.SUBMISSION_FAILED) {
						task.onSubmissionRejection();
					} else {
						try {
							task.onCompletion(this.result.get());
						} catch (Throwable tx) {
							task.onError(tx);
						}
					}
					return true;
				} else {
					FastIO.get().printlnError(_internal);
					return false;
				}

			}

		}
	}

	@Deprecated
	public Task<T> toLegacyTask() {
		return new Task<T>(this.callback) {

			@Override
			public T run() {
				FutureTask.this.run();
				return FutureTask.this.result.get();
			}

		};
	}

	@Override
	public List<AsyncTaskCallback<?>> callbacks() {
		return Lists.of(this.callback.toArray(new AsyncTaskCallback[this.callback.size()]));
	}

	@Override
	public void signalQueued() {
		this.status.lazySet(TaskStatus.QUEUED);

		if (!this.callback.isEmpty()) {
			if (ENABLE_OFFLOADED_ASYNC) {
				this.callback.iterator().forEachRemaining((_o) -> {
					if (_o == null)
						return;
					final AsyncTaskCallback<T> o = _o;
					ExecutorManager.get().autoAsyncSubmit(ENABLE_ASYNC_EVENT_SUBMISSION, () -> {
						try {
							o.onQueued();
						} catch (Throwable ec) {
							ec.printStackTrace(System.err);
							System.err.flush();
						}
					});
				});
			} else {
				this.callback.iterator().forEachRemaining((o) -> {
					if (o == null)
						return;
					try {
						o.onQueued();
					} catch (Throwable ec) {
						ec.printStackTrace(System.err);
						System.err.flush();
					}
				});
			}
		}
	}

	@Override
	public T waitFor(final long amount, final TimeUnit unit) {
		if (amount < 0 || unit == null)
			throw new IllegalArgumentException("amount < 0!");

		if ((USE_SEMAPHORE_EXPERIMENT ? this.completed() : this.countDownLatch.isCompleted())) {

			final TaskStatus _status = this.status.get();

			if (_status == TaskStatus.FAILED && USE_NEW_FAIL_LOGIC) {
				throw new RuntimeException(this.cachedError.get());
			} else if (_status == TaskStatus.SUBMISSION_FAILED && USE_NEW_FAIL_LOGIC) {
				throw new RuntimeException("Submission was rejected.");
			} else {
				return this.result.get();
			}

		} else {

			try {
				if (USE_SEMAPHORE_EXPERIMENT) {
					if (!this.sem.tryAcquire(1, amount, unit)) {
						if (this.t.get() != null)
							this.t.get().interrupt();
						throw new TaskTimedOutException(toString() + ": Timed out at waitFor(" + amount + ", " + unit + ")!", this.e);
					}
				} else {
					if (this.countDownLatch.await(amount, unit)) {
						if (this.t.get() != null)
							this.t.get().interrupt();
						throw new TaskTimedOutException(toString() + ": Timed out at waitFor(" + amount + ", " + unit + ")!", this.e);
					}
				}
			} catch (InterruptedException e) {
				throw new RuntimeException(e);
			}

			final TaskStatus _status = this.status.get();

			if (_status == TaskStatus.FAILED && USE_NEW_FAIL_LOGIC) {
				throw new RuntimeException(this.cachedError.get());
			} else if (_status == TaskStatus.SUBMISSION_FAILED && USE_NEW_FAIL_LOGIC) {
				throw new RuntimeException("Submission was rejected.");
			} else {
				return this.result.get();
			}
		}
	}

	@Override
	public T run_st() {
		this.run();
		return this.result.get();
	}

	@Override
	public Throwable getException() {
		return this.cachedError.get();
	}

	@Override
	public String toString() {
		return toString(false);
	}

	@Override
	public String toString(final boolean disableResultPrinting) {
		return "FutureTask[hash=" + MathUtil.toHex(super.hashCode()) + ",state=" + this.status + ",result="
				+ (disableResultPrinting ? "[OMITTED]" : this.result) + ",cachedError=" + this.cachedError + ",cdl=" + this.countDownLatch + "]";
	}

	@Override
	public boolean isRunning() {
		return this.status.get() == TaskStatus.RUNNING;
	}

	@Override
	public void rejectSubmission() {
		this.status.lazySet(TaskStatus.SUBMISSION_FAILED);

		if (USE_SEMAPHORE_EXPERIMENT) {
			this.sem.release(Integer.MAX_VALUE);
		} else {
			this.countDownLatch.countDown();
		}

		if (!this.callback.isEmpty()) {
			/* FIXME: Should we use offloaded-async here? */
			this.callback.iterator().forEachRemaining((o) -> {
				if (o == null)
					return;
				try {
					o.onSubmissionRejection();
				} catch (Throwable ec) {
					ec.printStackTrace(System.err);
					System.err.flush();
				}
			});
		}
	}

	/* FT 1.1 */

	@Override
	public boolean requestCancel() {
		if (this.t.get() == null)
			return false;

		try {
			this.t.get().notifyAll();
			this.t.get().interrupt();
			return true;
		} catch (Throwable ex) {
			return false;
		}
	}

	@Override
	public Object getFlag(final int flag) {
		switch (flag) {
			case FT_SUPPORTS_TRACES:
			case FT_SUPPORTS_POLLING:
				return FT_TRUE;
			default:
				return FT_NOT_SUPPORTED;
		}
	}

	@Override
	public int getFlags() {
		return FT_SUPPORTS_TRACES | FT_SUPPORTS_POLLING;
	}

	@Override
	public boolean requestReset() {
		this.t.lazySet(null);
		this.status.lazySet(TaskStatus.UNSUBMITTED);
		this.result.lazySet(null);
		this.cachedError.lazySet(null);
		this.callback.clear();

		if (USE_SEMAPHORE_EXPERIMENT) {
			this.sem.reset();
		} else {
			this.countDownLatch.countDownThenReset();
		}
		return true;
	}

	@Override
	public Throwable getTrace() {
		return this.e;
	}

	@Override
	public boolean hasTrace() {
		return this.e != null;
	}

	@Override
	public boolean hasExecutingThread() {
		return this.t.get() != null;
	}

	@Override
	public Thread getExecutingThread() {
		return this.t.get();
	}

	/* FT 1.1: Polling */

	@Override
	public T pollUntilCompletion(final int polling_algorithm) {
		switch (polling_algorithm) {
			case FT_POLLING_IMMEDIATE: {
				while (this.isRunning()) {
					try {
						Thread.sleep(1L);
					} catch (InterruptedException e) {
						Thread.interrupted();
					}
				}
				break;
			}
			case FT_POLLING_NEW_SCALING: {
				// TODO: Implement and create an algorithm for FT_POLLING_NEW_SCALING.
			}
			default:
			case FT_POLLING_LEGACY_SCALING: {
				final int limit = 8;
				int waitingInt = limit;
				final int s = (int) (FT_POLLING_LEGACY_SCALING_WAITING_INTERVAL / limit);
				while (this.isRunning()) {
					if (waitingInt <= 0) {
						waitingInt = limit + 1;
					}
					try {
						Thread.sleep((long) Math.ceil(s / waitingInt));
					} catch (InterruptedException e) {
						Thread.interrupted();
					}
					waitingInt -= 1;
				}
				break;
			}
		}

		if (this.status.get() == TaskStatus.FAILED && USE_NEW_FAIL_LOGIC)
			throw new RuntimeException(this.cachedError.get());
		else if (this.status.get() == TaskStatus.SUBMISSION_FAILED && USE_NEW_FAIL_LOGIC)
			throw new RuntimeException("Submission was rejected.");
		else
			return this.result.get();
	}

	@Override
	public T pollFor(final long amount, final TimeUnit unit, final int polling_algorithm) {
		final Timer time = new Timer();
		time.start(ExecutorManager.get().executor());
		final long am = unit.toMillis(amount);
		switch (polling_algorithm) {
			case FT_POLLING_IMMEDIATE: {
				while (this.isRunning() && !time.hasReached(am)) {
					try {
						Thread.sleep(1L);
					} catch (InterruptedException e) {
						Thread.interrupted();
					}
				}
				break;
			}
			case FT_POLLING_NEW_SCALING: {
				// TODO: Implement and create an algorithm for FT_POLLING_NEW_SCALING.
			}
			default:
			case FT_POLLING_LEGACY_SCALING: {
				final int limit = 8;
				int waitingInt = limit;
				final int s = (int) (FT_POLLING_LEGACY_SCALING_WAITING_INTERVAL / limit);
				while (this.isRunning() && !time.hasReached(am)) {
					if (waitingInt <= 0) {
						waitingInt = limit + 1;
					}
					try {
						Thread.sleep((long) Math.ceil(s / waitingInt));
					} catch (InterruptedException e) {
						Thread.interrupted();
					}
					waitingInt -= 1;
				}
				break;
			}
		}
		time.stop();

		if (this.status.get() == TaskStatus.FAILED && USE_NEW_FAIL_LOGIC)
			throw new RuntimeException(this.cachedError.get());
		else if (this.status.get() == TaskStatus.SUBMISSION_FAILED && USE_NEW_FAIL_LOGIC)
			throw new RuntimeException("Submission was rejected.");
		else
			return this.result.get();
	}

	/* Debugging */

	public static <T> Executable<T> dummy(final TaskStatus status) {
		Variables.requireNonNull(status, "status");

		final FutureTask<T> ft = new FutureTask<T>() {

			@Override
			public T execute() {
				throw new UnsupportedOperationException("Cannot execute a dummy object!");
			}
		};
		ft.status.set(status);

		return ft;
	}
}
